package main

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/gob"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go"
	"github.com/apache/rocketmq-client-go/consumer"
	"github.com/apache/rocketmq-client-go/primitive"
	"github.com/apache/rocketmq-client-go/rlog"
	"github.com/garyburd/redigo/redis"
	_ "github.com/go-sql-driver/mysql"
	"hash/crc32"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

const (
	userName = "root"
	password = "admin"
	ip = "120.79.162.72"
	port = "3307"
	dbName = "BusData"
)
var DB *sql.DB


type BusStatus struct {
	BusID 			string `json:"busID"`
	BusPlate 		string `json:"busPlate"`
	BusState 		string `json:"busState"`
	CompCode 		string `json:"compCode"`
	CurrentRuns 	string `json:"currentRus"`
	DateTime 		string `json:"dateTime"`
	GpsCode 		string `json:"gpsCode"`
	GpsDate 		string `json:"gpsDate"`
	GpsDirection 	string `json:"gpsDirection"`
	GpsMileage 		string `json:"gpsMileage"`
	GpsSpeed 		string `json:"gpsSpeed"`
	PlanRuns 		string `json:"planRuns"`
	PlanTime 		string `json:"planTime"`
	RoadID 			string `json:"roadID"`
	RoadName 		string `json:"roadName"`
	RoadType 		string `json:"roadType"`
	SiteId 			string `json:"siteId"`
	SiteName 		string `json:"siteName"`
	SiteNo 			string `json:"siteNo"`
	SiteState 		string `json:"siteState"`

}
type RoadSite struct {
	CompCode  string `json:"compCode"`
	GpsCode   string `json:"gpsCode"`
	RoadID    string `json:"roadID"`
	RoadName  string `json:"roadName"`
	RoadType  string `json:"roadType"`
	SiteId 	  string `json:"siteId"`
	SiteName  string `json:"siteName"`
	SiteNo    int `json:"siteNo"`
}
type BusInfo struct {
	BusId    string ` json:"busId"`
	BusPlate string ` json:"busPlate"`
	CompCode string ` json:"compCode"`
	RoadID   string ` json:"roadID"`
	RoadName string ` json:"roadName"`
}
type RoadInfo struct {
	ChargeType 		string `json:"chargeType"`
	CompCode 		string `json:"compCode"`
	DownFirstTime 	string `json:"downFirstTime"`
	DownKilo 		string `json:"downKilo"`
	DownLastTime 	string `json:"downLastTime"`
	LineFare 		string `json:"lineFare"`
	RoadID   		string `json:"roadID"`
	RoadName 		string `json:"roadName"`
	UpFirstTime 	string `json:"upFirstTime"`
	UpKilo 			string `json:"upKilo"`
	UpLastTime 		string `json:"upLastTime"`
}


type BusStatusMessage struct {
	Msg   	string `json:"msg"`
	Data 	[]BusStatus `json:"data"`
	Status 	string `json:"status"`
}
type RoadSiteMessage struct {
	Msg   	string `json:"msg"`
	Data 	[]RoadSite `json:"data"`
	Status 	string `json:"status"`
}
type RoadInfoMessage struct {
	Msg   	string `json:"msg"`
	Data 	[]RoadInfo `json:"data"`
	Status 	string `json:"status"`
}
type BusInfoMessage struct {
	Msg   	string `json:"msg"`
	Data 	[]BusInfo `json:"data"`
	Status 	string `json:"status"`
}

type BusStatusApi struct {
	RoadId string `json:"roadId"`
	RoadName string `json:"roadName"`
	CurrentSites []CurrentSite `json:"currentSites"`
}
type CurrentSite struct {
	SiteId string `json:"siteId"`
	SiteName string `json:"siteName"`
	BusList []BusPlate `json:"busList"`
}
type BusPlate struct {
	LicensePlate string `json:"licensePlate"`
}

type RoadSiteInfoApi struct {
	RoadId string `json:"roadId"`
	RoadName string `json:"direction"`
	RoadType string `json:"roadType"`
	FirstTime string ` json:"beginTime"`
	LastTime string `json:"endTime"`
	List []SiteLocation `json:"list"`
}
type SiteLocation struct {
	SiteId string `json:"siteId"`
	SiteName string `json:"siteName"`
	SiteNo int `json:"siteNo"`
	Lng string `json:"lng"`
	Lat string `json:"lat"`
}





func main ()  {
	stop := make(chan int)
	rlog.SetLogLevel("error")
	go startConsumer()
	go startWebServer()
	<- stop

}



func getBusStatus(w http.ResponseWriter, r *http.Request){
	w.Header().Set("Access-Control-Allow-Origin", "*")             //允许访问所有域
	w.Header().Add("Access-Control-Allow-Headers", "Content-Type") //header的类型
	w.Header().Set("content-type", "application/json")             //返回数据格式是json

	r.ParseForm()
	roadName := r.Form["roadName"][0]
	roadType := r.Form["roadType"][0]

	searchSiteNoKey := "RoadSite_" + roadName + "_" + roadType

	conn := redisConn()
	res, _ := redis.Ints(conn.Do("HKEYS", searchSiteNoKey))
	searchRoadInfoKey := "RoadInfo_" + roadName
	var roadInfo RoadInfo
	roadStr, _ := redis.String(conn.Do("GET", searchRoadInfoKey))
	err := json.Unmarshal([]byte(roadStr), &roadInfo)
	if err != nil {
		fmt.Println(err.Error())
	}

	var currentSites []CurrentSite
	for i := range res{ /*res为SiteNo集合*/
		var currentSite CurrentSite

		siteInfoStr, _ := redis.String(conn.Do("HGET", searchSiteNoKey, res[i]))

		var siteInfo RoadSite
		err := json.Unmarshal([]byte(siteInfoStr), &siteInfo)
		if err != nil  {
			fmt.Println(err.Error())
		}

		currentSite.SiteId = siteInfo.SiteId
		currentSite.SiteName = siteInfo.SiteName

		var busPlateList []BusPlate
		//fmt.Println("SELECT BusPlate FROM BusState WHERE RoadName = ? AND RoadType = ? AND SiteNo = ?", roadName+"路", roadType, res[i])

		format := "2006-01-02 15:04:05"
		now := time.Now()
		m, err := time.ParseDuration("-10m")
		if err != nil {
			fmt.Println(err.Error())
		}
		nowBefore := now.Add(m).Format(format) //现在时间的5分钟前

		rows, err := DB.Query("SELECT BusPlate FROM BusState WHERE RoadName = ? AND RoadType = ? AND SiteNo = ? AND GpsDate > ?", roadName+"路", roadType, res[i], nowBefore)
		if err != nil {
			fmt.Println("Query error", err.Error())
		}
		for rows.Next() {
			var busPlate BusPlate
			err := rows.Scan(&busPlate.LicensePlate)
			if err != nil {
				fmt.Println("Get bus plate error", err.Error())
			}
			busPlateList = append(busPlateList, busPlate)
		}
		rows.Close()
		currentSite.BusList = busPlateList

		currentSites = append(currentSites, currentSite)
	}


	var busStatusApi BusStatusApi
	busStatusApi.RoadName = roadInfo.RoadName
	busStatusApi.RoadId = roadInfo.RoadID
	busStatusApi.CurrentSites = currentSites
	returnJson, _ := json.Marshal(busStatusApi)

	fmt.Fprintln(w, string(returnJson))
	defer conn.Close()

}

func getRoadSite(w http.ResponseWriter, r *http.Request){
	w.Header().Set("Access-Control-Allow-Origin", "*")             //允许访问所有域
	w.Header().Add("Access-Control-Allow-Headers", "Content-Type") //header的类型
	w.Header().Set("content-type", "application/json")             //返回数据格式是json


	r.ParseForm()
	roadName := r.Form["roadName"][0]
	roadType := r.Form["roadType"][0]
	conn := redisConn()

	searchRoadSiteListKey := "RoadSite_"+roadName+"_"+roadType
	roadInfoStr, _ := redis.String(conn.Do("GET", "RoadInfo_"+roadName))
	roadSiteList, _ := redis.Strings(conn.Do("HKEYS", searchRoadSiteListKey))

	var siteLocationList []SiteLocation
	for i := range roadSiteList{
		var siteLocation SiteLocation
		var tempRoadSite RoadSite

		siteLocationStr, _ := redis.String(conn.Do("HGET", searchRoadSiteListKey, roadSiteList[i]))
		err := json.Unmarshal([]byte(siteLocationStr), &tempRoadSite)
		if err != nil {
			fmt.Println(err.Error())
			continue
		}

		gpsCode := strings.Split(tempRoadSite.GpsCode, ",")
		siteLocation.SiteName = tempRoadSite.SiteName
		siteLocation.SiteId = tempRoadSite.SiteId
		siteLocation.SiteNo = tempRoadSite.SiteNo
		siteLocation.Lat = gpsCode[0]
		siteLocation.Lng = gpsCode[1]

		siteLocationList = append(siteLocationList, siteLocation)
	}

	var roadSiteInfoApi RoadSiteInfoApi
	var roadInfo RoadInfo
	json.Unmarshal([]byte(roadInfoStr), &roadInfo)
	roadSiteInfoApi.RoadId = roadInfo.RoadID
	roadSiteInfoApi.RoadName = roadInfo.RoadName
	roadSiteInfoApi.RoadType = roadType

	if roadType == "0"{
		roadSiteInfoApi.FirstTime = roadInfo.UpFirstTime
		roadSiteInfoApi.LastTime = roadInfo.UpLastTime
	}else if roadType == "1"{
		roadSiteInfoApi.FirstTime = roadInfo.DownFirstTime
		roadSiteInfoApi.LastTime = roadInfo.DownLastTime
	}
	roadSiteInfoApi.List = siteLocationList

	roadSiteInfoJson, _ := json.Marshal(roadSiteInfoApi)


	defer conn.Close()
	fmt.Fprintln(w, string(roadSiteInfoJson))
}

func webApi(w http.ResponseWriter, r *http.Request){
	r.ParseForm()
	fmt.Fprintf(w, "Hello webApi!")
}



func startConsumer(){
	initMysql()

	insertBusStatusStmt, _ := DB.Prepare("INSERT INTO BusState (SiteId,BusId,GpsMileage,SiteName,SiteNo,BusPlate,PlanRuns,RoadName,PlanTime,RoadID,GpsSpeed,DateTime,CompCode,SiteState,GpsDate,GpsDirection,RoadType,BusState,GpsCode,CurrentRuns) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
	updateBusStatusStmt, _ := DB.Prepare("UPDATE BusState SET GpsDate = ?, GpsCode = ?, GpsSpeed = ?, GpsMileage = ?, RoadType = ?, SiteId = ?, SiteNo = ?, SiteName = ?, SiteState = ?, BusState = ? WHERE  BusPlate = ? AND RoadID = ?")



	c, _ := rocketmq.NewPushConsumer(
		consumer.WithGroupName("Consumber"),
		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"120.79.162.72:9876"})),
	)

	//BusStatus operation
	err := c.Subscribe("BusStatus", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		redisConn := redisConn()
		fmt.Println("Have one bus status message ... ")
		for i := range msgs {
			var busStatusMessage BusStatusMessage
			if err := json.Unmarshal(msgs[i].Body[:], &busStatusMessage); err != nil {
				fmt.Println("Error" + err.Error())
			}
			data := busStatusMessage.Data
			format := "2006-01-02 15:04:05"
			loc, _ := time.LoadLocation("Local")
			now := time.Now()
			m, err := time.ParseDuration("-2m")
			if err != nil {
				fmt.Println(err.Error())
			}
			nowBefore5m := now.Add(m) //现在时间的5分钟前
			//fmt.Println(nowBefore5m.Format(format))
			//timeToShow, _ := time.Parse(format, "2020-06-17 19:00:00")


			for j := range data {
				state := data[j]

				dataTime, err := time.ParseInLocation(format, state.GpsDate, loc)
				if err != nil {
					fmt.Println()
				}

				if dataTime.After(nowBefore5m) {
					if state.SiteState == "0" {
						switch selectBusByBusPlate(state,redisConn) {
						case 0:
							insertBusStatus(state, insertBusStatusStmt)
						case 1:
							upDateBusStatus(state, updateBusStatusStmt)
						case -1:
							fmt.Println("Exec error")
						default:
							fmt.Println("Unknown state")
						}
					}
				}

			}
		}

		defer redisConn.Close()
		fmt.Println("Data done")
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		fmt.Print(err.Error())
	}


	//RoadSite operation
	err = c.Subscribe("RoadSite", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		redisConn := redisConn()
		for i := range msgs {
			var roadSiteMessage RoadSiteMessage
			if err := json.Unmarshal(msgs[i].Body[:], &roadSiteMessage); err != nil {
				fmt.Println("Error" + err.Error())
			}

			data := roadSiteMessage.Data
			//fmt.Println("RoadSite:",len(data))
			for j := range data{
				//var site = data[j]
				roadName := strings.Replace(data[j].RoadName, "路", "", 1) + "_"
				roadType := data[j].RoadType
				hashcode := "RoadSite_" + roadName + roadType
				content, _ := json.Marshal(data[j])
				exists, _ := redis.Bool(redisConn.Do("HEXISTS", hashcode, data[j].SiteNo))
				if !exists {
					redisConn.Do("HSET", hashcode, data[j].SiteNo, content)
				}


			}
		}
		defer redisConn.Close()
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {

		fmt.Print(err.Error())
	}


	//BusInfo operation
	err = c.Subscribe("BusInfo", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		redisConn := redisConn()
		for i := range msgs {
			var busInfoMessage BusInfoMessage
			if err := json.Unmarshal(msgs[i].Body[:], &busInfoMessage); err != nil {
				fmt.Println("Error" + err.Error())
			}

			data := busInfoMessage.Data
			//fmt.Println("BusInfo:",len(data))
			for j := range data{
				var bus = data[j]
				roadName := strings.Replace(data[j].RoadName, "路", "", 1) + "_"
				hashcode := "BusInfo_" + roadName + strconv.Itoa(hashcode(bus))
				content, _ := json.Marshal(data[j])
				exists, _ := redis.Bool(redisConn.Do("EXISTS", hashcode))
				if !exists {
					redisConn.Do("set", hashcode, content)
				}

			}
		}
		defer redisConn.Close()
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {

		fmt.Print(err.Error())
	}


	//RoadInfo operation
	err = c.Subscribe("RoadInfo", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		redisConn := redisConn()
		for i := range msgs {
			var roadInfoMessage RoadInfoMessage
			if err := json.Unmarshal(msgs[i].Body[:], &roadInfoMessage); err != nil {
				fmt.Println("Error" + err.Error())
			}

			data := roadInfoMessage.Data
			//fmt.Println("RoadInfo:",len(data))
			for j := range data{
				var road = data[j]
				roadName := strings.Replace(road.RoadName, "路", "", 1)
				hashcode := "RoadInfo_" + roadName
				content, _ := json.Marshal(road)
				exists, _ := redis.Bool(redisConn.Do("EXISTS", hashcode))
				if !exists {
					redisConn.Do("set", hashcode, content)
				}
			}

		}
		defer redisConn.Close()
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {

		fmt.Print(err.Error())
	}
	//start
	err = c.Start()
	if err != nil {
		fmt.Print(err.Error())
		os.Exit(-1)
	}
	//
	//err = c.Shutdown()
	//if err != nil {
	//	fmt.Println("shutdown Consumer error: %s", err.Error())
	//
	//}
}

func startWebServer(){
	http.HandleFunc("/", webApi) // 设置访问的路由
	http.HandleFunc("/getRoadSite", getRoadSite)
	http.HandleFunc("/getBusStatus", getBusStatus)
	err := http.ListenAndServe(":7788", nil) // 设置监听的端口
	if err != nil {
		log.Fatal("ListenAndServe: ", err)
	}
}



func redisConn() redis.Conn{
	conn, err := redis.Dial("tcp", "120.79.162.72:4836")
	if err != nil {
		fmt.Println(err.Error())
	}
	return conn
}

func initMysql() {
	path := strings.Join([]string{userName, ":", password, "@tcp(",ip, ":", port, ")/", dbName, "?charset=utf8"}, "")
	DB, _ = sql.Open("mysql", path)
	if err := DB.Ping(); err != nil{
		fmt.Println("opon database fail")
		return
	}
}

/*Get statement */



func selectBusBySite(roadName string,roadType string, siteNo int) []BusPlate{
	rows, err := DB.Query("SELECT BusPlate FROM BusStatus WHERE RoadName = ? AND RoadType = ? AND SiteNo = ?", roadName, roadType, siteNo)
	if err != nil {
		fmt.Println("查询出错")
	}
	var busPlates []BusPlate
	for rows.Next() {
		var busPlate BusPlate
		rows.Scan(&busPlate.LicensePlate)
		busPlates = append(busPlates, busPlate)
	}
	return busPlates
}
func selectBusByBusPlate(status BusStatus, redisConn redis.Conn) int8{
	queryStr := "BusPlate_" + strings.Replace(status.RoadName, "路", "", 1)
	var flag int8
	isBusPlateExists, err := redis.Bool(redisConn.Do("HEXISTS", queryStr, status.BusPlate))
	if err != nil {
		fmt.Println("redis query error" + err.Error())
		flag = -1
	}
	if isBusPlateExists{
		flag = 1
	}else {
		redisConn.Do("HSET", queryStr, status.BusPlate, status.SiteNo)
		flag = 0
	}
	return flag
}
func upDateBusStatus(busStatus BusStatus, stmt *sql.Stmt) bool{
	tx, err := DB.Begin()
	if err != nil{
		fmt.Println("Update tx fail", err.Error())
	}

	res, err := stmt.Exec(busStatus.GpsDate, busStatus.GpsCode, busStatus.GpsSpeed,busStatus.GpsMileage, busStatus.RoadType, busStatus.SiteId, busStatus.SiteNo, busStatus.SiteName,busStatus.SiteState, busStatus.BusState, busStatus.BusPlate, busStatus.RoadID)
	if err != nil {
		fmt.Println("exec fail", err.Error())
		return false
	}
	tx.Commit()
	res.LastInsertId()
	return true
}
func insertBusStatus(bus BusStatus, stmt *sql.Stmt) bool{
	res, err := stmt.Exec(bus.SiteId,bus.BusID,bus.GpsMileage,bus.SiteName,bus.SiteNo,bus.BusPlate,bus.PlanRuns,bus.RoadName,bus.PlanTime,bus.RoadID,bus.GpsSpeed,bus.DateTime,bus.CompCode,bus.SiteState,bus.GpsDate,bus.GpsDirection,bus.RoadType,bus.BusState,bus.GpsCode,bus.CurrentRuns)
	if err != nil {
		fmt.Println("Exec fail", err.Error())
		return false
	}
	res.LastInsertId()
	return true
}




func hashcode(site interface{}) int {
	var buf bytes.Buffer
	enc := gob.NewEncoder(&buf)
	err :=enc.Encode(site)
	if err != nil {
		fmt.Println(err.Error())
	}
	v := int(crc32.ChecksumIEEE(buf.Bytes()))
	if v >= 0 {
		return v
	}
	if -v >= 0 {
		return -v
	}
	// v == MinInt
	return 0

}
